package com.zb.netty.chat.client;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;

import com.zb.netty.chat.client.handler.RpcResponseMessageHandler;
import com.zb.netty.chat.message.RpcRequestMessage;
import com.zb.netty.chat.protocal.MessageCodecSharable;
import com.zb.netty.chat.protocal.ProtocolFrameDecoder;
import com.zb.netty.chat.protocal.SequenceIdGenerator;
import com.zb.netty.chat.server.service.HelloService;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;

/**
 * rpc客户端
 **/
public class RpcClientV2 {
    private static Channel channel = null;
    private static final Object LOCK = new Object();

    public static void main(String[] args) {
        HelloService service = getService(HelloService.class);
        System.out.println("result1=" + service.sayHello("zhangsan"));
        System.out.println("result2=" + service.sayHello("lisi"));
    }

    public static Channel getChannel() {
        if (null != channel) {
            return channel;
        }
        synchronized (LOCK) {
            if (null != channel) {
                return channel;
            }
            initChannel();
        }
        return channel;
    }

    /**
     * 通过代理模式，发送请求消息
     * @param serviceClass 代理的接口
     * @param <T> 代理类型
     * @return 代理对象
     */
    public static <T> T getService(Class<T> serviceClass) {
        // 获取serviceClass的类加载器
        ClassLoader classLoader = serviceClass.getClassLoader();
        // 获取serviceClass的实现接口
        Class<?>[] interfaces = {serviceClass};
        // 创建增强的对象
        Object proxy = Proxy.newProxyInstance(classLoader, interfaces, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 封装rpc请求消息对象RpcRequestMessage
                int nextId = SequenceIdGenerator.nextId();
                RpcRequestMessage rpcRequestMessage = new RpcRequestMessage(nextId,
                        serviceClass.getName(),
                        method.getName(),
                        method.getReturnType(),
                        method.getParameterTypes(),
                        args);
                // 发送消息
                getChannel().writeAndFlush(rpcRequestMessage);
                // 创建Promise，用于获取NIO线程中的返回结果。获取的过程是异步的,参数为channel的eventLoop（指定异步接收结果线程，promise中addListener中线程）
                DefaultPromise<Object> defaultPromise = new DefaultPromise<>(getChannel().eventLoop());
                // 将Promise放入Map中
                RpcResponseMessageHandler.PROMISES.put(nextId, defaultPromise);
                // 等待被放入Promise中结果
                defaultPromise.await();
                if (defaultPromise.isSuccess()) {
                    // 调用方法成功，返回方法执行结果
                    return defaultPromise.getNow();
                } else {
                    // 调用方法失败，抛出异常
                    throw new RuntimeException(defaultPromise.cause());
                }
            }
        });
        return (T) proxy;
    }

    /**
     * 初始化channel
     */
    private static void initChannel() {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
        // RPC响应消息处理器
        RpcResponseMessageHandler RPC_RESPONSE_HANDLER = new RpcResponseMessageHandler();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ch.pipeline().addLast(new ProtocolFrameDecoder());
                ch.pipeline().addLast(LOGGING_HANDLER);
                ch.pipeline().addLast(MESSAGE_CODEC);
                ch.pipeline().addLast(RPC_RESPONSE_HANDLER);
            }
        });
        try {
            channel = bootstrap.connect(new InetSocketAddress("localhost", 8089))
                    .sync().channel();
            channel.closeFuture().addListener(future -> eventLoopGroup.shutdownGracefully());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
